RSMS worker — phased implementation plan
Status: Active — rsms-worker-function uses a self-hosted Python process (queue_worker.py) that polls Azure Storage Queue and runs the same engine + rsms_results_storage pipeline. Azure Functions is not the production host (Python Function Apps on Windows VMs were not viable); rsms-api-fastapi enqueues jobs (202) when RSMS_WORKER_QUEUE_NAME and AZURE_STORAGE_CONNECTION_STRING are set.
Design reference: rsms-worker-function/function-rework-04272026/rework-readme.md (queue polling, visibility timeout, JSON vs base64 decode, poison handling). That folder is design-only; the POC script there is not production code—implementation is merged into the parent worker package.
Parent context: rsms-results-table-storage-rework-plan, rsms-results-storage-design (blob layout, Table plume contract).
Goal: rsms-api-fastapi enqueues a job ( riverbasin_id, scenario_id, optional metadata) to Azure Storage Queue. A long-running worker dequeues, runs the RSMS pipeline, and writes simulation results to Azure Table Storage (PlumeByTime / PlumeByMile) and JSON blobs (edge_timeseries.json, mass_balance.json) via rsms_results_storage, without requiring a developer notebook on the execution path.
1. References (normative)
| Doc / artifact | Role |
|---|---|
| rsms-results-storage-design | Blob containers (rsms-results vs riverbasin), RiverMileIndex.csv path, minified JSON |
| rsms-results-table-storage-rework-plan | Worker ↔ storage integration phases |
rsms-worker-function/function-rework-04272026/rework-readme.md | Queue worker behavior (poll interval, encoding, poison queue) |
latest-notebooks-melissa/ notebooks | Historical step-order reference — engine_workflow.py implements the authoritative sequence |
rsms-worker-function/rsms_results_storage/ | Writer pipeline (owned under the worker) |
2. Locked decisions (summary)
- Trigger: Azure Storage Queue (API sends messages; worker receives with visibility timeout and
dequeue_countfor retries). Not Table polling for job dispatch. - Message body: JSON with
riverbasin_idandscenario_id(same contract asrun_rsms_cli; optionaljob_id,triggeredBy, etc.). Plain JSON in code; support base64-wrapped payloads for tools that encode that way (see design readme). - River mile index: download
{riverbasin_id}/RiverMileIndex.csvfrom blob containerriverbasin(orRSMS_RIVERBASIN_BLOB_CONTAINER), not SQL Server at runtime. - Config:
local.settings.jsonValuesand/or.env(team standard); templatersms-worker-function/local.settings.example.json. Queue name and connection string via env (e.g. reuseAZURE_STORAGE_CONNECTION_STRINGwhen the queue is in the same account). - Results write: once, after the full workflow succeeds (no partial upload on failed step).
- Local testing: Thin CLI (or
python -m …) that accepts the same JSON payload and invokes the shared handler without the queue—parity with integration tests and developer workflows. - Tests: Lightweight unit tests where they pay off (e.g. message parsing); avoid a large suite until the queue + API contract is stable.
3. Out of scope
- Azure Functions runtime as the primary production host for this worker (
function_app.py/azure-functionsdependency removed).
4. Phases
Phases are sequential: each phase should be demoable before the next. A single large PR is possible if the team uses this section as an internal checklist; smaller PRs per phase reduce risk.
Historical W1–W3 work (blob index, engine parity, writer) is largely done. W4 (queue worker, run_rsms_handlers, API enqueue) is implemented in code; the tables below remain the checklist / reference.
Phase W1 — Package layout, config, and river mile index from blob
| Task | Notes |
|---|---|
Relocate rsms_results_storage | Canonical copy under rsms-worker-function (single source of truth). |
| Dependencies | requirements.txt: azure-data-tables, azure-storage-blob, azure-storage-queue (worker + API enqueue); azure-functions removed. |
| Settings | AZURE_STORAGE_CONNECTION_STRING, RSMS_RESULTS_BLOB_CONTAINER, RSMS_RIVERBASIN_BLOB_CONTAINER, EXE_DIR, SQL/API vars for NFQ as today. Add queue settings (queue name, optional poison queue name, poll interval, visibility timeout, max dequeue). |
Load RiverMileIndex.csv | Unchanged: blob → pandas.read_csv → validate columns. |
Exit criteria: Worker code can load the index CSV from Azure Blob for a given riverbasin_id using connection string + container settings.
Phase W2 — Engine workflow parity
| Task | Notes |
|---|---|
| Map legacy script order → modules | engine_workflow, exe_runner, etc. (melissa/ notebooks = historical sketches only). |
| Scenario source | Load Scenarios from Azure Table as today; queue message carries ids only unless you extend the contract. |
| Subprocess | Absolute paths, EXE_DIR, capture stdout/stderr and exit codes; fail fast on missing .TIM / .PLT. |
| Logging | Structured logs per step; no secrets in logs. |
Exit criteria: End-to-end run on a dev machine produces engine outputs and (when enabled) uploads results for a test scenario.
Phase W3 — Results writer (Table + JSON blobs)
| Task | Notes |
|---|---|
| Call pipeline | After successful CXPLT + parsers, write_scenario_results_to_azure with local run folder, connection string, ids, concentration_tolerance. |
| Idempotency | Document upsert behavior for replay. |
| Smoke test | PlumeByTime / PlumeByMile and JSON blobs under {riverbasin_id}/{scenario_id}/. |
Exit criteria: rsms-api-fastapi results endpoints work for that scenario after a successful worker run.
Phase W4 — Queue worker, API enqueue, and operations
| Task | Notes |
|---|---|
| Factor handlers | run_rsms_from_body in run_rsms_handlers.py ( function_app.py removed). |
| Worker loop | queue_worker.py: visibility timeout, queue_payload.decode_queue_message_content, poison queue, drain-then-repoll. |
| Thin CLI | run_rsms_cli.py (JSON file or stdin). |
| FastAPI | enqueue_rsms_run when RSMS_WORKER_QUEUE_NAME + AZURE_STORAGE_CONNECTION_STRING set (202); returns 503 if the queue is not configured. |
| Timeouts | RSMS_WORKER_VISIBILITY_TIMEOUT_SEC (default 7200); renewal not yet implemented. |
| Errors | Worker logs failures; engine_workflow updates scenario Failed on engine errors. |
| Deployment | rsms-worker-windows-service-nssm — RSMS VM + NSSM; see rsms-worker-function/README.md — not Python Function App. |
| CI | Optional; tests/test_queue_payload.py covers message decode. |
Exit criteria: Operations can run the queue worker on a Windows-capable host with a checklist; API triggers runs via queue; troubleshooting does not require notebook access.
5. Dependency graph
W1 (blob CSV + package)
↓
W2 (engine parity)
↓
W3 (writer)
↓
W4 (queue worker + API enqueue + ops)
Open backlog (visibility renewal, ops nits): rsms-master-tasklist
RSMS worker — phased implementation plan.